#include "NetIO.h"

Reactor_Impl::Reactor_Impl( void )
	:exit_( 0 )
{
}

Reactor_Impl::~Reactor_Impl( void )
{
}

int Reactor_Impl::async_read( Socket handle, DataBlock *data_block, void *act )
{
	Operation *op =
	    ObjectCache<Operation>::alloc( );
	op->reset( this );
	op->data_block( data_block );
	op->act( act );
	op->handle( handle );
	op->optype( OperationType::READ );

	if( data_block == 0 || handle == INVALID_HANDLE )
		return result_queue_.enqueue_tail( op );

	if( exit_ )
		return result_queue_.enqueue_tail( op );

	if( this->start_op( handle, op ) == -1 )
		return result_queue_.enqueue_tail( op );
	return 0;
}

int Reactor_Impl::async_write( Socket handle, DataBlock *data_block, void *act )
{
	if( exit_ )
		return -1;

	Operation *op =
	    ObjectCache<Operation>::alloc( );
	op->reset( this );
	op->data_block( data_block );
	op->act( act );
	op->handle( handle );
	op->optype( OperationType::WRITE );

	if( data_block == 0 || handle == INVALID_HANDLE )
		return result_queue_.enqueue_tail( op );

	if( exit_ )
		return result_queue_.enqueue_tail( op );

	if( this->start_op( handle, op ) == -1 )
		return result_queue_.enqueue_tail( op );

	return 0;
}

int Reactor_Impl::async_connect( const string &ip, short port, void *act )
{
	if( exit_ )
		return -1;

	Operation *op =
	    ObjectCache<Operation>::alloc( );
	Socket &new_handle = op->handle( );

	sockaddr_in addr = {0};
	socklen_t len = sizeof( addr );
	addr.sin_family = AF_INET;
	addr.sin_port = htons( port );
	addr.sin_addr.s_addr = inet_addr( ip.c_str( ) );
	new_handle( ) = socket( AF_INET, SOCK_STREAM, 0 );
	SockOps::set_non_blocking( new_handle );

	op->reset( this );
	op->optype( OperationType::CONNECT );
	op->act( act );

	if( connect( new_handle, ( sockaddr* )( &addr ), len ) == 0 )
	{
		result_queue_.enqueue_tail( op );
		return 0;
	}

	register_handler( new_handle );
	if( start_op( new_handle, op ) == -1 )
		return result_queue_.enqueue_tail( op );

	return 0;
}

int Reactor_Impl::run( )
{
	while( run_once( ) == 0 )
	{

	}

	return 0;
}

int Reactor_Impl::async_accept( Socket listen_handle, void *act )
{
	if( exit_ )
		return -1;

	Operation *op =
	    ObjectCache<Operation>::alloc( );
	op->reset( this );
	op->act( act );
	op->handle( listen_handle );
	op->optype( OperationType::ACCEPT );

	if( listen_handle == INVALID_HANDLE )
		return result_queue_.enqueue_tail( op );

	if( exit_ )
		return result_queue_.enqueue_tail( op );

	if( this->start_op( listen_handle, op ) == -1 )
		return result_queue_.enqueue_tail( op );

	return 0;
}

int Reactor_Impl::handle_all_complete_op( )
{
	result_queue_.concat_queue( &complete_operations_ );
	return 0;
}

int Reactor_Impl::operation_result( Operation_Result &result )
{
	Operation *op;
	result_queue_.dequeue_head( op );

	if( !op )
		return -1;

	result.optype_ = op->optype( );
	result.errcode_ = op->errcode( );
	result.data_block_ = op->data_block( );
	result.act_ = op->act( );
	result.bytes_transferred_ = op->bytes_transferred( );
	result.handle_ = op->handle( );
	ObjectCache<Operation>::free( op );
	return 0;
}

void Reactor_Impl::close( )
{
	Guard<Synch_Mutex> guard( mutex_ );
	exit_ = true;
	guard.release( );

	cancel_all_op( );
	interrupter_.interrupt( );
}

int Reactor_Impl::register_handler( Socket &handle )
{
	TRACE_FUNC;

	if( !handle.is_open( ) )
		return -1;

	Guard<Synch_Mutex> guard( handle_map_mutex_ );
	handle_map::iterator iter = handle_map_.find( handle );
	if( iter != handle_map_.end( ) )
	{
		handle.handle_data( iter->second );
		return 0;
	}
	Handle_Data *data = Handle_Data::make( );
	handle_map_.insert( handle, data );
	handle.handle_data( data );

	return register_handler_i( handle );
}

void Reactor_Impl::close_handle( Socket &handle )
{
	
}

void Reactor_Impl::cancel_all_op( )
{
	Guard<Synch_Mutex> guard_map( handle_map_mutex_ );
	for( handle_map::iterator iter = handle_map_.begin( ); iter != handle_map_.end( ); ++iter )
	{
		Guard<Synch_Mutex> guard_handle( iter->second->mutex_ );
		for( int i = 0; i < OperationType::CONNECT; ++i )
			result_queue_.concat( &iter->second->operations_[ i ] );
	}
}

